package com.atguigu.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.bean.TableProcess;
import com.atguigu.realtime.commont.Constant;
import com.atguigu.realtime.util.JdbcUtil;
import com.atguigu.realtime.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/11/24 10:12
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection phoenixConn;
    private ValueState<Boolean> tableCreatedState;
    private ValueState<String> upsertSqlState;
    private Jedis redisClient;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取键控状态  初始化Phoenix jdbc连接
        
        phoenixConn = JdbcUtil.getPhoenixConnection(Constant.PHOENIX_URL);
        
        tableCreatedState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("tableCreatedState", Boolean.class));
        
        upsertSqlState = getRuntimeContext().getState(new ValueStateDescriptor<String>("upsertSqlState", String.class));
        
        redisClient = RedisUtil.getRedisClient();
        
    }
    
    @Override
    public void close() throws Exception {
        if (phoenixConn != null) {
            phoenixConn.close();
        }
    }
    
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> t,
                       Context context) throws Exception {
        // 实现写入的具体逻辑
        // 1. 检测表是否存在, 不存在先建表
        checkTable(t);
        // 2. 把数据写入到对应的表中
        writeToPhoenix(t);
        
        // 3. 如果这个数据是更新, 去查看redis'是否存在这个维度, 如果存在就需要更新redis'中的维度信息
        updateRedis(t);
    }
    
    private void updateRedis(Tuple2<JSONObject, TableProcess> t) {
        JSONObject data = t.f0;
        TableProcess tp = t.f1;
        
        // 如果这次是更新
        if ("update".equals(t.f1.getOperate_type())) {
            // 去redis判断key是否存在   注意大小写的问题
            String key = t.f1.getSink_table() + ":" + data.getLong("id");
            
            JSONObject upper = new JSONObject();
            // 优雅: 更新数据
            if (redisClient.exists(key)) {
                for (Map.Entry<String, Object> entry : data.entrySet()) {
                    upper.put(entry.getKey().toUpperCase(), entry.getValue());
                }
                redisClient.set(key, upper.toJSONString());
            }
            // 粗暴
            //redisClient.del(key);
        }
        
    }
    
    private void writeToPhoenix(Tuple2<JSONObject, TableProcess> t) throws SQLException, IOException {
        JSONObject data = t.f0;
        TableProcess tp = t.f1;
        
        if (upsertSqlState.value() == null) {
            // 拼接sql语句
            // upsert into user(name, id, age)values(?,?,?)
            StringBuilder insertSql = new StringBuilder();
            insertSql
                .append("upsert into ")
                .append(tp.getSink_table())
                .append("(")
                // 字段名
                .append(tp.getSink_columns())
                .append(")values(")
                // 占位符
                .append(tp.getSink_columns().replaceAll("[^,]+", "?"))  // 所有的非逗号用问号替换
                .append(")");
            
            upsertSqlState.update(insertSql.toString());
            System.out.println("插入语句: " + insertSql.toString());
        }
        
        PreparedStatement ps = phoenixConn.prepareStatement(upsertSqlState.value());
        
        // 给sql语句中的占位符进行赋值
        String[] columns = tp.getSink_columns().split(",");
        for (int i = 0; i < columns.length; i++) {
            // 根据列名去 data中取出对应的值
            String column = columns[i];
            String value = data.getString(column);
            ps.setString(i + 1, value);  // 站位符的计数是基于1
        }
        
        ps.execute();
        phoenixConn.commit();
        ps.close();
        
    }
    
    private void checkTable(Tuple2<JSONObject, TableProcess> t) throws SQLException, IOException {
        
        if (tableCreatedState.value() == null) {
            TableProcess tp = t.f1;
            
            // 执行一个建表语句
            // create table if not exists user(id varchar, name varchar, age varchar, constraint pk_name primary key(id))
            StringBuilder sql = new StringBuilder();
            sql
                .append("create table if not exists ")
                .append(tp.getSink_table())
                .append("(")
                .append(tp.getSink_columns().replaceAll(",", " varchar,"))
                .append(" varchar, constraint pk_name primary key(")
                .append(tp.getSink_pk() == null ? "id" : tp.getSink_pk())
                .append("))")
                .append(tp.getSink_extend() == null ? "" : tp.getSink_extend());
            System.out.println("建表语句: " + sql.toString());
            // 拼接sql语句
            PreparedStatement ps = phoenixConn.prepareStatement(sql.toString());
            ps.execute();
            phoenixConn.commit();
            ps.close();
            
            tableCreatedState.update(true);
        }
    }
}
